package com.eelpo.framework.socket.session;

import java.net.SocketAddress;
import java.util.AbstractSet;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.ChannelGroupFuture;
import org.jboss.netty.channel.group.DefaultChannelGroupFuture;
import org.jboss.netty.util.internal.ConcurrentHashMap;

public class SocketSessionGroup extends AbstractSet<Channel> implements ChannelGroup {

	private final String name;
	private static final AtomicInteger nextId = new AtomicInteger();
	private final ConcurrentMap<Integer, SocketSession> sessions = new ConcurrentHashMap<Integer, SocketSession>();

	private final ChannelFutureListener remover = new ChannelFutureListener() {
		public void operationComplete(ChannelFuture future) throws Exception {
			remove(future.getChannel());
		}
	};

	public SocketSessionGroup() {
		this("group-0x" + Integer.toHexString(nextId.incrementAndGet()));
	}

	public SocketSessionGroup(String name) {
		if (name == null) {
			throw new NullPointerException("name");
		}
		this.name = name;
	}

	public void closeWithoutFuture() {
		for (SocketSession session : sessions.values()) {
			session.closeWithoutFuture();
		}
	}

	public void disconnectWithoutFuture() {
		for (SocketSession session : sessions.values()) {
			session.disconnectWithoutFuture();
		}
	}

	public void setInterestOpsWithoutFuture(int interestOps) {
		for (SocketSession session : sessions.values()) {
			session.setInterestOpsWithoutFuture(interestOps);
		}
	}

	public void setReadableWithoutFuture(boolean readable) {
		for (SocketSession session : sessions.values()) {
			session.setReadableWithoutFuture(readable);
		}
	}

	public void unbindWithoutFuture() {
		for (SocketSession session : sessions.values()) {
			session.unbindWithoutFuture();
		}
	}

	public void writeWithoutFuture(Object message) {
		if (message instanceof ChannelBuffer) {
			ChannelBuffer buf = (ChannelBuffer) message;
			for (SocketSession session : sessions.values()) {
				session.writeWithoutFuture(buf.duplicate());
			}
		} else {
			for (SocketSession session : sessions.values()) {
				session.writeWithoutFuture(message);
			}
		}
	}

	public void writeWithoutFuture(Object message, SocketAddress remoteAddress) {
		if (message instanceof ChannelBuffer) {
			ChannelBuffer buf = (ChannelBuffer) message;
			for (SocketSession session : sessions.values()) {
				session.writeWithoutFuture(buf.duplicate(), remoteAddress);
			}
		} else {
			for (SocketSession session : sessions.values()) {
				session.writeWithoutFuture(message, remoteAddress);
			}
		}
	}

	public String getName() {
		return name;
	}

	@Override
	public boolean isEmpty() {
		return sessions.isEmpty();
	}

	@Override
	public int size() {
		return sessions.size();
	}

	public Channel find(Integer id) {
		return sessions.get(id);
	}

	@Override
	public boolean contains(Object o) {
		if (o instanceof Integer) {
			return sessions.containsKey(o);
		} else if (o instanceof Channel) {
			return sessions.containsKey(((Channel) o).getId());
		} else {
			return false;
		}
	}

	@Override
	public boolean add(Channel channel) {
		boolean added = sessions.putIfAbsent(channel.getId(), (SocketSession) channel) == null;
		if (added) {
			channel.getCloseFuture().addListener(remover);
		}
		return added;
	}

	@Override
	public boolean remove(Object o) {
		Channel c = null;
		if (o instanceof Integer) {
			c = sessions.remove(o);
		} else if (o instanceof Channel) {
			c = sessions.remove(((Channel) o).getId());
		}

		if (c == null) {
			return false;
		}

		c.getCloseFuture().removeListener(remover);
		return true;
	}

	@Override
	public void clear() {
		sessions.clear();
	}

	@Override
	public Iterator<Channel> iterator() {
		Collection<SocketSession> session = sessions.values();
		Channel[] channels = new Channel[session.size()];
		session.toArray(channels);
		return Arrays.asList(channels).iterator();
	}

	@Override
	public Object[] toArray() {
		return sessions.values().toArray();
	}

	@Override
	public <T> T[] toArray(T[] a) {
		return sessions.values().toArray(a);
	}

	public ChannelGroupFuture close() {
		List<ChannelFuture> futures = new ArrayList<ChannelFuture>(size());
		for (Channel c : sessions.values()) {
			futures.add(c.close());
		}
		return new DefaultChannelGroupFuture(this, futures);
	}

	public ChannelGroupFuture disconnect() {
		List<ChannelFuture> futures = new ArrayList<ChannelFuture>(size());
		for (Channel c : sessions.values()) {
			futures.add(c.disconnect());
		}
		return new DefaultChannelGroupFuture(this, futures);
	}

	public ChannelGroupFuture setInterestOps(int interestOps) {
		List<ChannelFuture> futures = new ArrayList<ChannelFuture>(size());
		for (Channel c : sessions.values()) {
			futures.add(c.setInterestOps(interestOps));
		}

		return new DefaultChannelGroupFuture(this, futures);
	}

	public ChannelGroupFuture setReadable(boolean readable) {
		List<ChannelFuture> futures = new ArrayList<ChannelFuture>(size());
		for (Channel c : sessions.values()) {
			futures.add(c.setReadable(readable));
		}

		return new DefaultChannelGroupFuture(this, futures);
	}

	public ChannelGroupFuture unbind() {
		List<ChannelFuture> futures = new ArrayList<ChannelFuture>(size());
		for (Channel c : sessions.values()) {
			futures.add(c.unbind());
		}

		return new DefaultChannelGroupFuture(this, futures);
	}

	public ChannelGroupFuture write(Object message) {
		List<ChannelFuture> futures = new ArrayList<ChannelFuture>(size());
		if (message instanceof ChannelBuffer) {
			ChannelBuffer buf = (ChannelBuffer) message;
			for (Channel c : sessions.values()) {
				futures.add(c.write(buf.duplicate()));
			}
		} else {
			for (Channel c : sessions.values()) {
				futures.add(c.write(message));
			}
		}
		return new DefaultChannelGroupFuture(this, futures);
	}

	public ChannelGroupFuture write(Object message, SocketAddress remoteAddress) {
		List<ChannelFuture> futures = new ArrayList<ChannelFuture>(size());
		if (message instanceof ChannelBuffer) {
			ChannelBuffer buf = (ChannelBuffer) message;
			for (Channel c : sessions.values()) {
				futures.add(c.write(buf.duplicate(), remoteAddress));
			}
		} else {
			for (Channel c : sessions.values()) {
				futures.add(c.write(message, remoteAddress));
			}
		}
		return new DefaultChannelGroupFuture(this, futures);
	}

	@Override
	public int hashCode() {
		return System.identityHashCode(this);
	}

	@Override
	public boolean equals(Object o) {
		return this == o;
	}

	public int compareTo(ChannelGroup o) {
		int v = getName().compareTo(o.getName());
		if (v != 0) {
			return v;
		}

		return System.identityHashCode(this) - System.identityHashCode(o);
	}

	@Override
	public String toString() {
		return getClass().getSimpleName() + "(name: " + getName() + ", size: " + size() + ')';
	}
}
